package com.jiajia.study.chapter01.nio;

import com.jiajia.study.utils.ByteBufferUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author zjiajia
 * @date 2024/7/5 14:11
 */
@Slf4j
public class TestServerV4 {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");


        Selector boss = Selector.open();

        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);

        SelectionKey serverSocketSelectionKey = serverSocketChannel.register(boss, 0, null);
        serverSocketSelectionKey.interestOps(SelectionKey.OP_ACCEPT);

        serverSocketChannel.bind(new InetSocketAddress(8888));
        int workerLength = 2;
        // 初始化 worker
        Worker[] workers = new Worker[workerLength];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker("worker-" + i);
        }
        AtomicInteger number = new AtomicInteger(0);

        while(true){
            boss.select();
            Set<SelectionKey> selectionKeys = boss.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while(iterator.hasNext()){
                SelectionKey selectionKey = iterator.next();
                if(selectionKey.isAcceptable()){
                    iterator.remove();
                    serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    log.debug("connected...{}", socketChannel.getRemoteAddress());
                    // 2. 关联 selector
                    log.debug("before register...{}", socketChannel.getRemoteAddress());
                    workers[number.getAndIncrement() % workers.length].register(socketChannel);
                    log.debug("after reggister...{}",socketChannel.getRemoteAddress());
                }

            }
        }


    }
}
@Slf4j
class Worker implements Runnable{
    private String name;
    private Thread thread;
    private Selector selector;
    private volatile boolean start = false;

    public Worker(String name){
        this.name = name;
    }

    public void register(SocketChannel socketChannel) throws IOException {
        if(!start){
            selector = Selector.open();
            // 开启一个线程
            thread = new Thread(this,name);
            thread.start();
            start = true;
        }
        /**
         * 唤醒 selector.select 的阻塞
         *
         * 避免出现：
         *   一直阻塞在 selector,然后新的监听一直卡在这
         */
        selector.wakeup();
        // channel 的 读写事件监听给这个线程
        socketChannel.register(selector, SelectionKey.OP_READ,null);
    }


    @Override
    public void run() {
        while(true){
            try {
                /**
                 * 监听事件的发生
                 */
                selector.select();

                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                if(iterator.hasNext()){
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    try{
                        if(selectionKey.isReadable()){
                            ByteBuffer byteBuffer = ByteBuffer.allocate(32);
                            log.debug("当前处理线程：{},read...{}",Thread.currentThread().getName(), socketChannel.getRemoteAddress());
                            int read = socketChannel.read(byteBuffer);
                            if(read == -1 ){
                                selectionKey.cancel();
                            }
                            // 切换成读模式
                            byteBuffer.flip();
                            // 数据读取
                            ByteBufferUtil.debugAll(byteBuffer);
                            System.out.println(StandardCharsets.UTF_8.decode(byteBuffer));
                        }else if(selectionKey.isWritable()){
                            // todo 写数据
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                        /**
                         * 异常关闭，取消监听
                         */
                        selectionKey.cancel();
                    }


                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
